package com.lds.memory.kafka.consumer.parallel;

import com.lds.memory.kafka.consumer.AbstractConsumer;
import com.lds.memory.kafka.partition.Partition;
import com.lds.memory.kafka.topic.Topic;

import java.lang.reflect.Constructor;
import java.util.List;

public class ConsumeTopicTask {
    private final Topic topic;
    private final Class<? extends AbstractConsumer> consumerClass;

    public ConsumeTopicTask(Topic topic, Class<? extends AbstractConsumer> consumerClass) {
        this.topic = topic;
        this.consumerClass = consumerClass;
    }

    public void start() {
        try {
            List<Partition> partitions = topic.getPartitions();
            for (int i = 0, size = partitions.size(); i < size; i++) {
                Partition partition = partitions.get(i);

                Constructor<? extends AbstractConsumer> constructor = consumerClass.getConstructor(Partition.class);
                AbstractConsumer consumer = constructor.newInstance(partition);
                ConsumerTask consumerTask = new ConsumerTask(consumer);

                partition.bind(consumerTask);

                new Thread(consumerTask, "memory-kafka-thread" + i).start();
            }

            // 消费者线程启动成功后,立即启动topic
            // 没有消费者时,启动topic毫无意义
            topic.start();
        } catch (Exception e) {}
    }

}
